/*********************************************************************************
* @file         socket_tcp_mqtt.c
* @ingroup      userapp
* @brief        使用MQTT API实现与MQTT服务器的通信
* @attention
* @note         
* @author       WP              
**********************************************************************************/
#include "zos.h"
#include "xy_api.h"
#include "xy_utils.h"
#include "xy_demo.h"
#include "MQTTClient.h"
#include "MQTTTimer.h"

/*authentication information*/
#define MQTT_ClientID   "57950b7ee9f842aab21a16b51c327568"
#define MQTT_UserName   "wp"
#define MQTT_PassWord   ""

/*MQTT message param*/
#define MQTT_Topic_Sub  "Sub/device_control"
#define MQTT_Topic_Pub  "Pub/device_control"
#define MQTT_CleanSession   1
#define MQTT_QOS            1

/*will information*/
#define MQTT_Will_Flag      1
#define MQTT_Will_Topic     "device_control"
#define MQTT_Will_Message   "will message"
#define MQTT_Will_Qos       1
#define MQTT_Will_Retained  0

#define MQTT_Command_Timeout    5000
#define MQTT_KeepLive           600
#define MQTT_Sendbuf_Size       200
#define MQTT_Readbuf_Size       200

/*SELECT ERROR CODE*/
#define SELETE_TIMEOUT          -2
#define SELETE_NET_ERR          -1
#define PORT                    10162

/*DOWNLINK RECV  BUF*/
#define PACKET_HEAD             "add "
#define MAX_RECV_LEN            5
#define MAX_PACKET_SIZE         200
#define SEND_PACKET_CNT         30


extern char grebuf[UARTLEN];
extern double gcalmsg[10];

static osThreadId_t g_mqttemq_handle = NULL;
static osThreadId_t g_reconectemq_handle = NULL;
static osThreadId_t g_mqttdownlink_handle = NULL;
static osThreadId_t g_deepsleep_handle = NULL;
static MQTTClient *pMQTT_emq_client={NULL};
static zos_sem_t entry_deepsleep_sem = NULL;
static zos_sem_t reconnect_sem = NULL;
static int packetcnt = 0;


struct Demo_Options
{
    char* host;         /* server IP or domain name */
    int port;
    char* proxy_host;
    int proxy_port;
    int MQTTVersion;
} network_options =
{
    "64.69.43.237",//ngrok
    10702,//
    "mqtt.ctwing.cn",
    1885,
    4,
};

void mqtt_user_defined_process(MessageData *md)
{
    char *packet = NULL;
    char *dst = NULL;
    int len=0,i=0;
    MQTTMessage* m = md->message;
    zos_printf("[MQTTemq]payload was %s\r\n",m->payload);

    packet = zos_malloc(sizeof(char)*MAX_RECV_LEN);
    memset(packet,0,sizeof(char)*MAX_RECV_LEN);
    if((dst=strstr(m->payload,PACKET_HEAD))!=NULL)
    {
        len = strlen(PACKET_HEAD);
        while(dst[i+len] != ' ')
        {
            packet[i] = dst[i+len];
            i++;
            if(i>MAX_RECV_LEN)
                break;
        }
    }

    packetcnt = atoi(packet);
    if(packetcnt>MAX_PACKET_SIZE)
        packetcnt = MAX_PACKET_SIZE;
    
    zos_printf("packetcnt %d\r\n",packetcnt);
    zos_free(packet);
    packet = NULL;
}

void mqtt_downlink_buf_recv(MQTTClient *c)
{
    int ret = -1;
    int timeout = 2000;
    struct timeval tv;
    fd_set read_fds, exceptfds;

    FD_ZERO(&read_fds);
    FD_ZERO(&exceptfds);

    FD_SET(c->ipstack->my_socket, &read_fds);
    FD_SET(c->ipstack->my_socket, &exceptfds);

    tv.tv_sec  = timeout / 1000;
    tv.tv_usec = (timeout % 1000) * 1000;
    ret = select(c->ipstack->my_socket+1, &read_fds, NULL, &exceptfds, &tv);

    if(ret == 0)
    {
        zos_printf("[MQTTemq] select timeout, no new data\r\n");
        return;
    }
    else if(ret < 0)
    {
        close(c->ipstack->my_socket);
        c->ipstack->my_socket = -1;
        zos_printf("[MQTTemq] select error ret=%d,err %d\r\n", ret, errno);
        return ;
    }

    if(FD_ISSET(c->ipstack->my_socket, &exceptfds))
    {
        close(c->ipstack->my_socket);
        c->ipstack->my_socket = -1;
        return ;
    }

    if(FD_ISSET(c->ipstack->my_socket, &read_fds))
    {
        ret = recv(c->ipstack->my_socket, c->readbuf, c->readbuf_size, 0);

        if (ret < 0)
        {
            if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)
            {
                zos_printf("[MQTTemq] no data available for now\r\n");
                return ;
            }
            else
            {
                zos_printf("[MQTTemq] error accured when recv: 0x%x\r\n", errno);
                close(c->ipstack->my_socket);
                c->ipstack->my_socket =-1;
                return ;
            }
        }
        else if(ret == 0)
        {
            zos_printf("[MQTTemq] socket was closed by peer\r\n");
            close(c->ipstack->my_socket);
            c->ipstack->my_socket =-1;
            return ;
        }
        else
        {
            zos_printf("[MQTTemq] packet arrived\r\n");
            xy_mqtt_downlink_process(pMQTT_emq_client,mqtt_user_defined_process,1); //处理下行数据
        }

    }

    return ;
} 

void mqtt_downlink_message_process(void)
{   
    while(1)
    {
        if(pMQTT_emq_client->ipstack->my_socket < 0 || pMQTT_emq_client->close_sock == 1)
            return ;
        keepalive(pMQTT_emq_client);
        mqtt_downlink_buf_recv(pMQTT_emq_client);
    }
}

void mqtt_downlink_message_rev(void)
{
    mqtt_downlink_message_process();
    g_mqttdownlink_handle = NULL;
    osThreadExit();
}

void mqtt_downlink_message_process_thread_create()
{
    if(g_mqttdownlink_handle != NULL)
        return ;
    g_mqttdownlink_handle = osThreadNew((osThreadFunc_t)(mqtt_downlink_message_rev),NULL,"mqtt_downlink_message_rev",0x1000,osPriorityNormal);
}

void mqtt_connect_param_init(MQTTPacket_connectData *data)
{
    data->MQTTVersion = network_options.MQTTVersion;
    data->clientID.cstring = MQTT_ClientID;
    data->username.cstring = MQTT_UserName;
    data->password.cstring = MQTT_PassWord;
    data->keepAliveInterval= MQTT_KeepLive;
    data->cleansession     = MQTT_CleanSession;

    data->willFlag               = MQTT_Will_Flag;
    data->will.message.cstring   = MQTT_Will_Message;
    data->will.qos               = MQTT_Will_Qos;
    data->will.retained          = MQTT_Will_Retained;
    data->will.topicName.cstring = MQTT_Will_Topic;
    return;
}

void mqtt_publish_param_init(MQTTMessage* pubmsg, char *buf, uint8_t len, uint8_t qos)
{
    pubmsg->payload = buf;              // What to 浮点数发送有问题？
    pubmsg->payloadlen = len;          //数据的完整性
    pubmsg->qos = qos;
    pubmsg->retained = 0;
    pubmsg->dup = 0;
    return;
}

void free_demo_memory(MQTTClient* c)
{
    int i;
	if(c == NULL)
		return;
    for (i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
    {
    	if(c->messageHandlers[i].topicFilter)
    	{
    		xy_free((void *)(c->messageHandlers[i].topicFilter));
			c->messageHandlers[i].topicFilter = NULL;
		}
    }
        
    if(c->ipstack)
        xy_free(c->ipstack);
    if(c->buf)
        xy_free(c->buf);
    if(c->readbuf)
        xy_free(c->readbuf);
    
    xy_free(c);
	c = NULL;
	return;
}

static void grebuf_to_msg(struct attitude_msg_raw *msg,char *grebuf)
{
	msg->ahead = grebuf[0];
    msg->aflag = grebuf[1];
    msg->axl = grebuf[2];
    msg->axh = grebuf[3];
    msg->ayl = grebuf[4];
    msg->ayh = grebuf[5];
    msg->azl = grebuf[6];
    msg->azh = grebuf[7];
    msg->atl = grebuf[8];
    msg->ath = grebuf[9];
    msg->wflag = grebuf[12];
    msg->wxl = grebuf[13];
    msg->wxh = grebuf[14];
    msg->wyl = grebuf[15];
    msg->wyh = grebuf[16];
    msg->wzl = grebuf[17];
    msg->wzh = grebuf[18];
    msg->wtl = grebuf[19];
    msg->wth = grebuf[20];
    msg->wsum = grebuf[21];
	msg->anglehead = grebuf[22];
    msg->angleflag = grebuf[23];
    msg->rollxl = grebuf[24];
    msg->rollxh = grebuf[25];
    msg->pitchyl = grebuf[26];
    msg->pitchyh = grebuf[27];
    msg->yawzl = grebuf[28];
    msg->yawzh = grebuf[29];
    msg->angletl = grebuf[30];
    msg->angleth = grebuf[31];
    msg->anglesum = grebuf[32];
}

static int mqtt_tcp_emq_dashboard(void)
{
    int rc = 0, cnt = 0, disconnectcnt = 0, tempcnt=0;
    char *sendbuf = xy_zalloc(MQTT_Sendbuf_Size);
    char *readbuf = xy_zalloc(MQTT_Readbuf_Size);
    char *emq_topic_sub = xy_zalloc(sizeof(MQTT_Topic_Sub)+1);
    char *emq_topic_pub = xy_zalloc(sizeof(MQTT_Topic_Pub)+1);
    pMQTT_emq_client = xy_zalloc(sizeof(MQTTClient));
    pMQTT_emq_client->ipstack = xy_zalloc(sizeof(Network));

    strncpy(emq_topic_sub,MQTT_Topic_Sub,strlen(MQTT_Topic_Sub));
    strncpy(emq_topic_pub,MQTT_Topic_Pub,strlen(MQTT_Topic_Pub));

    if(entry_deepsleep_sem == NULL)
        entry_deepsleep_sem = zos_sem_create(0xFFFF, 0);
    if(reconnect_sem == NULL)
        reconnect_sem = zos_sem_create(0xFFFF, 0);

    /*network init*/
    NetworkInit(pMQTT_emq_client->ipstack);
    network_options.port = PORT;
	zos_printf("host:%s port:%d\r\n", network_options.host,network_options.port);
    NetworkConnect(pMQTT_emq_client->ipstack,network_options.host,network_options.port);
    MQTTClientInit((MQTTClient*)(pMQTT_emq_client),pMQTT_emq_client->ipstack,(unsigned int)(MQTT_Command_Timeout),
    (unsigned char*)(sendbuf), (size_t)(MQTT_Sendbuf_Size),(unsigned char*)(readbuf),(size_t)(MQTT_Readbuf_Size));

    /*Create thread to receive downlink message asynchronously*/
    mqtt_downlink_message_process_thread_create();

    /*Init MQTT connect message Parameters*/
    MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
    mqtt_connect_param_init(&data);

    /*send MQTT connect message*/
    rc = xy_mqtt_connect(pMQTT_emq_client,&data,1);
    xy_assert(SUCCESS==rc);

    /*send MQTT subscribe message*/
    rc = xy_mqtt_subscribe(pMQTT_emq_client, emq_topic_sub, MQTT_QOS,1);

    MQTTMessage pubmsg = {0};
    struct attitude_msg_raw *msg = NULL;
    msg = zos_malloc(sizeof(struct attitude_msg_raw));
    
    while(1)
    {
        msg->power = -1;
        grebuf_to_msg(msg,grebuf);
        /*Init MQTT publish message Parameters*/
        if(nb_vbat_low_check()==ZOS_EOK)
            msg->power = 0;
        msg->packetcnt = cnt;
        mqtt_publish_param_init(&pubmsg, (char*)msg, sizeof(struct attitude_msg_raw), MQTT_QOS);

        /*send MQTT publish message*/
        rc = xy_mqtt_publish(pMQTT_emq_client,emq_topic_pub,&pubmsg,1);
        if(rc!=0)
            disconnectcnt++;
        else
        {
            disconnectcnt=0;
            zos_printf("publish msg cnt=%d\r\n",cnt);
        } 

        if(disconnectcnt>10)
        {
            disconnectcnt = 0;
            if(reconnect_sem != NULL)
                zos_sem_post(reconnect_sem);
            break;
        }
        (packetcnt==0)?(tempcnt=SEND_PACKET_CNT):(tempcnt=SEND_PACKET_CNT+packetcnt);
        zos_printf("tempcnt=%d\r\n",tempcnt);
        if(cnt++ >= tempcnt)
        {
            cnt = 0;
            disconnectcnt = 0;
            if(entry_deepsleep_sem != NULL)
            {
                send_cmd_sleep_set();
                zos_sem_post(entry_deepsleep_sem);
            } 
        }
        osDelay(1000);
    }

    /*send MQTT unsubscribe message*/
    rc = xy_mqtt_unsubscribe(pMQTT_emq_client,emq_topic_sub,1);

    /*send MQTT disconnect message */
    rc = xy_mqtt_disconnect(pMQTT_emq_client,1);

    /*close receive downlink thread */
    pMQTT_emq_client->close_sock = 1;

    osDelay(3000);
    xy_assert(g_mqttdownlink_handle==NULL);
    /* free MQTT client*/
    zos_free(msg);
    free_demo_memory(pMQTT_emq_client);

    return SUCCESS;
}

void deepsleep_rtc_callback(void *param)
{
    ;
}

void mqtt_tcp_emq_dashboard_task(void)
{
    //wait PDP active
    if(xy_wait_tcpip_ok(2*60)==XY_ERR)
        xy_assert(0);
    mqtt_tcp_emq_dashboard();
    g_mqttemq_handle = NULL;
    osThreadExit();
}

void reconnect_emq_task(void)
{
    while(1)
    {
        zos_sem_wait(reconnect_sem,ZOS_WAITING_FOREVER);
        mqtt_tcp_emq_dashboard_task();
        osDelay(2000);
    }

    g_reconectemq_handle = NULL;
    osThreadExit();
}

void deepsleep_task(void)
{
	struct nb_time_t time;
    //uint32_t srand = 0;

	osDelay(1000);
    if(xy_wait_tcpip_ok(2*60)==XY_ERR)              //使用zos_rtc_set_by_day，确保PAP联网
        xy_assert(0);

	osDelay(1000);
	if(zos_rtc_get_time(&time) == ZOS_EOK)          //随机sec
	{
        srand((time.tm_sec+time.tm_min));
	}
    //zos_rtc_timer_create(ZOS_RTC_TIMER_ID_1,60*62,deepsleep_rtc_callback,NULL);
    zos_rtc_set_by_day(ZOS_RTC_TIMER_ID_1, deepsleep_rtc_callback,NULL, 14, rand()%60, 20,0);

    nb_work_lock();
    while(1)
    {
        zos_sem_wait(entry_deepsleep_sem,ZOS_WAITING_FOREVER);
        //nb_work_unlock();
        nb_fast_power_off();
    }

    g_deepsleep_handle = NULL;
    osThreadExit();
}   

void mqtt_tcp_emq_dashboard_init(void)
{
    if(g_mqttemq_handle != NULL && g_reconectemq_handle!=NULL && g_deepsleep_handle!=NULL)
        return ;
    g_mqttemq_handle = osThreadNew ((osThreadFunc_t)(mqtt_tcp_emq_dashboard_task),NULL,"mqtt_tcp_emq_dashboard_task",0x1000,osPriorityNormal1);
    g_reconectemq_handle = osThreadNew ((osThreadFunc_t)(reconnect_emq_task),NULL,"reconnect_emq_task",0x400,osPriorityNormal1);
    g_deepsleep_handle = osThreadNew ((osThreadFunc_t)(deepsleep_task),NULL,"deepsleep_task",0x400,osPriorityNormal1);
}

